package com.swak.reactivex.cluster;

import java.util.Set;

import com.swak.loadbalance.LoadBalance;
import com.swak.loadbalance.LoadBalances;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.utils.Sets;

/**
 * Channels 管理所有的通道.
 * 
 * @author 618lf
 */
public class Channels {

	private final Set<ExchangeChannel> channels = Sets.newConcurrentSet();
	private final LoadBalance<ExchangeChannel> select = LoadBalances.RoundRobin.newLoadBalance();

	public Set<ExchangeChannel> channels() {
		return this.channels;
	}

	public Channels add(ExchangeChannel channel) {
		this.channels.add(channel);
		this.select.onRefresh(this.channels().stream().toList());
		return this;
	}

	public Channels remove(ExchangeChannel channel) {
		this.channels.remove(channel);
		this.select.onRefresh(this.channels().stream().toList());
		return this;
	}

	public boolean isEmpty() {
		return this.channels.isEmpty();
	}

	public int size() {
		return this.channels.size();
	}

	public ExchangeChannel select() {
		return this.select.select();
	}

	public void close() {
		channels.forEach(channel -> {
			try {
				channel.close();
			} catch (Exception e) {
			}
		});
		channels.clear();
		this.select.onRefresh(this.channels().stream().toList());
	}
}
